package com.jiaying.sync.kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;

/**
 * Kafka管理器
 * 负责Kafka消费者的初始化、配置和管理
 */
public class KafkaManager {
    private static final Logger logger = LoggerFactory.getLogger(KafkaManager.class);
    private static final String CONFIG_FILE = "application.properties";
    
    private KafkaConsumer<String, String> consumer;
    private AtomicBoolean running = new AtomicBoolean(false);
    
    /**
     * 构造函数
     */
    public KafkaManager() {
        initialize();
    }
    
    /**
     * 初始化Kafka消费者
     */
    private void initialize() {
        try {
            // 创建消费者配置
            Properties props = new Properties();
            // 从配置文件加载Kafka集群配置
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaBootstrapServers());
            props.put(ConsumerConfig.GROUP_ID_CONFIG, "mongo-cdc-group-" + System.currentTimeMillis());
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // 禁用自动提交
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从最早的消息开始消费
            
            // 性能优化配置
            props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "102400"); // 100KB，减少网络传输次数
            props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "500"); // 减少等待时间
            props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1000"); // 增加每次poll的最大记录数
            
            // 创建消费者实例
            this.consumer = new KafkaConsumer<>(props);
            
            logger.info("Kafka消费者初始化成功");
        } catch (Exception e) {
            logger.error("Kafka消费者初始化失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 获取Kafka集群配置
     * @return Kafka集群地址列表
     */
    private String getKafkaBootstrapServers() {
        Properties props = new Properties();
        try (java.io.InputStream input = getClass().getClassLoader().getResourceAsStream(CONFIG_FILE)) {
            if (input == null) {
                logger.error("无法找到配置文件: {}", CONFIG_FILE);
                return "192.168.5.125:9092"; // 默认值
            }
            props.load(input);
        } catch (java.io.IOException ex) {
            logger.error("加载配置文件失败: {}", ex.getMessage(), ex);
            return "192.168.5.125:9092"; // 默认值
        }
        
        // 获取Kafka集群配置，如果没有则使用默认值
        return props.getProperty("kafka.bootstrap.servers", "192.168.5.125:9092");
    }
    
    /**
     * 订阅主题
     * @param topicPattern 主题模式（正则表达式）
     * @param rebalanceListener 分区重平衡监听器
     */
    public void subscribe(String topicPattern, ConsumerRebalanceListener rebalanceListener) {
        if (consumer != null) {
            consumer.subscribe(Pattern.compile(topicPattern), rebalanceListener);
            logger.info("已订阅主题模式: {}", topicPattern);
        }
    }
    
    /**
     * 轮询消息
     * @param timeout 超时时间（毫秒）
     * @return 消费记录
     * @throws WakeupException 唤醒异常
     */
    public ConsumerRecords<String, String> poll(long timeout) throws WakeupException {
        if (consumer != null) {
            return consumer.poll(Duration.ofMillis(timeout));
        }
        return ConsumerRecords.empty();
    }
    
    /**
     * 唤醒消费者
     */
    public void wakeup() {
        if (consumer != null) {
            consumer.wakeup();
        }
    }
    
    /**
     * 提交偏移量
     */
    public void commitSync() {
        if (consumer != null) {
            consumer.commitSync();
        }
    }
    
    /**
     * 获取指定分区的当前位置
     * @param partition 分区
     * @return 当前位置
     */
    public long position(TopicPartition partition) {
        if (consumer != null) {
            return consumer.position(partition);
        }
        return -1;
    }
    
    /**
     * 获取分配给消费者的分区
     * @return 分区集合
     */
    public Set<TopicPartition> assignment() {
        if (consumer != null) {
            return consumer.assignment();
        }
        return Collections.emptySet();
    }
    
    /**
     * 设置消费者偏移量
     * @param partition 分区
     * @param offset 偏移量
     */
    public void seek(TopicPartition partition, long offset) {
        if (consumer != null) {
            consumer.seek(partition, offset);
        }
    }
    
    /**
     * 关闭消费者
     */
    public void close() {
        if (consumer != null) {
            try {
                consumer.close();
                logger.info("Kafka消费者已关闭");
            } catch (Exception e) {
                logger.error("关闭Kafka消费者时发生错误: {}", e.getMessage(), e);
            }
        }
    }
    
    /**
     * 检查消费者是否已初始化
     * @return 消费者是否已初始化
     */
    public boolean isInitialized() {
        return consumer != null;
    }
}